package com.shujia.sql

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{DataFrame, SparkSession}

object Demo3DSLAPI {
  def main(args: Array[String]): Unit = {

    /**
     * 在新版本的spark中，如果想要编写spark sql的话，需要使用新的spark入口类：SparkSession
     */
    val sparkSession: SparkSession = SparkSession.builder()
      .master("local")
      .appName("dsl语法api演示")
      .config("spark.sql.shuffle.partitions",1) //默认分区的数量是200个
      .getOrCreate()

    //导入Spark sql中所有的sql隐式转换函数
    import org.apache.spark.sql.functions._
    //导入另一个隐式转换，后面可以直接使用$函数引用字段进行处理
    import sparkSession.implicits._

    /**
     * DSL api
     */
    //新版本的读取方式，读取一个json数据,不需要手动指定列名
    //    val stuDF1: DataFrame = sparkSession.read
    //      .json("spark/data/students.json")

    //以前版本读取方式，灵活度要高一些
    val stuDF: DataFrame = sparkSession.read
      .format("json")
      .load("spark/data/students.json")

    //    stuDF2.show(100, truncate = false) //传入展示总条数,并完全显示数据内容
    /**
     * select 函数：选择数据【字段】，和纯sql语句中select意思基本是一样，在数据的前提上选择要留下的列
     *
     */
    //根据字段的名字选择要查询的字段
    //    stuDF.select("id","name","age").show(1)
    //    //根据字段的名字选择要查询的字段,selectExpr 可以传入表达式字符串形式
    //    stuDF.selectExpr("id","name","age","age + 1 as new_age").show()
    //使用隐式转换中的$函数将字段变成一个对象
    //    stuDF.select($"id",$"name",$"age").show(10)
    //使用对象做处理
    //    stuDF.select($"id",$"name",$"age" + 1  as "new_age").show(10)
    //可以在select中使用sql的函数
    //下面的操作等同于sql：select id,name,age+1 as new_age,substring(clazz,0,2) as km from lines;
    //    stuDF.select($"id",$"name",$"age" + 1  as "new_age",substring($"clazz",0,2) as "km").show(10)

    /**
     * where 函数：过滤数据
     */
    //直接将sql中where语句以字符串的形式传参
    //    stuDF.where("gender='女' and age=23").show()
    //使用$列对象的形式过滤
    // =!= 不等于
    // === 等于
    //    stuDF.where($"gender" === "女" and $"age" === 23).show()
    //    stuDF.where($"gender" =!= "男" and $"age" === 23).show()
    //过滤文科的学生
    //    stuDF.where(substring($"clazz", 0, 2) === "文科").show()

    /**
     * groupBy 分组函数
     * agg 聚合函数
     * 分组聚合要在一起使用
     * 分组聚合之后的结果DF中只会包含分组字段和聚合字段
     * select中无法出现不是分组的字段
     */
    //根据班级分组，求每个班级的人数和平均年龄
    //    stuDF.groupBy($"clazz")
    //      .agg(count($"clazz") as "number",round(avg($"age"),2) as "avg_age").show()

    /**
     * orderBy: 排序
     */
    //    stuDF.groupBy($"clazz")
    //          .agg(count($"clazz") as "number")
    //          .orderBy($"number").show()

    /**
     * join: 表关联
     */
    val scoreDF: DataFrame = sparkSession.read
      .format("csv")
      .option("sep", ",")
      .schema("id STRING,subject_id STRING,score INT")
      .load("spark/data/score.txt")
    //    scoreDF.show()

    //关联场景1：所关联的字段名字不一样的时候
    //    stuDF.join(scoreDF, $"id" === $"sid", "inner").show()
    //关联场景2：所关联的字段名字一样的时候
    //    stuDF.join(scoreDF,"id").show()

    /**
     * 开窗函数
     * 统计每个班级总分前3的学生
     *
     * 开窗不会改变总条数的，会以新增一列的形式加上开窗的结果
     * withColumn 新增一列
     */
    val joinStuAndScoreWithIDDF: DataFrame = stuDF.join(scoreDF, "id")
    joinStuAndScoreWithIDDF.groupBy($"id", $"clazz") //根据学号和班级一起分组
      .agg(sum($"score") as "sumScore") //计算总分
      .withColumn("rn", row_number() over Window.partitionBy($"clazz").orderBy($"sumScore".desc))
      //.select($"id", $"clazz", $"sumScore", row_number() over Window.partitionBy($"clazz").orderBy($"sumScore".desc) as "rn")
      .where($"rn" <= 3)
      .show()


  }
}
